#include "connection_manager_service.h"

#include "im/adapters/sock_message_outlet_adapter.h"

namespace imserver {

void UidConnServerMap::set(int64_t uid, const std::string& server_key) {
  LOG_INFO("UidConnServerMap::set --> uid:" << uid << " server_key:" << server_key);
  std::string cmd;
  if (server_key.find("sock_message_outlet_service") != std::string::npos) {
    cmd = "SET " + boost::lexical_cast<std::string>(uid) + ":sock_server_key" + " " + server_key + " EX 259200";
  } else if (server_key.find("comet_message_outlet_service") != std::string::npos) {
    cmd = "SET " + boost::lexical_cast<std::string>(uid) + ":comet_server_key" + " " + server_key + " EX 259200";
  } else {
    return;
  }
  redisReply * reply = NULL;
  bool res = false;
  for (int i = 0; i < 2; ++i) {
    reply = redisExecutor.Execute("redis server_key", cmd.c_str());
    if (reply != NULL) {
      if (reply->type == REDIS_REPLY_STATUS && strcasecmp(reply->str, "ok") == 0) {
        res = true;
      }
      freeReplyObject(reply);
      if (res) {
        break;
      }
    }
  }
  if (!res) {
    LOG_ERROR("UidConnServerMap::set --> redis set failed. uid:" << uid << " server_key:" << server_key);
  }
}

void UidConnServerMap::setValid(int64_t uid, const std::string& server_key) {
  LOG_INFO("UidConnServerMap::setValid --> uid:" << uid << " server_key:" << server_key);
  std::string cmd;
  if (server_key.find("sock_message_outlet_service") != std::string::npos) {
    cmd = "SET " + boost::lexical_cast<std::string>(uid) + ":sock_server_key" + " " + server_key + ":ok EX 259200";
  } else if (server_key.find("comet_message_outlet_service") != std::string::npos) {
    cmd = "SET " + boost::lexical_cast<std::string>(uid) + ":comet_server_key" + " " + server_key + ":ok EX 259200";
  } else {
    return;
  }
  redisReply * reply = NULL;
  bool res = false;
  for (int i = 0; i < 2; ++i) {
    reply = redisExecutor.Execute("redis server_key", cmd.c_str());
    if (reply != NULL) {
      if (reply->type == REDIS_REPLY_STATUS && strcasecmp(reply->str, "ok") == 0) {
        res = true;
      }
      freeReplyObject(reply);
      if (res) {
        break;
      }
    }
  }
  if (!res) {
    LOG_ERROR("UidConnServerMap::setValid --> redis set failed. uid:" << uid << " server_key:" << server_key);
  }
}

std::string UidConnServerMap::alloc(int64_t uid, const std::string& conn_type) {
  LOG_INFO("UidCometServerMap::create --> uid:" << uid);
  std::string version_path;
  if (conn_type == "sock") {
    version_path = "/im/sock_message_outlet_service/v1";
  } else if (conn_type == "comet") {
    version_path = "/im/comet_message_outlet_service/v1";
  } else {
    return "";
  }
  std::vector<std::string> shards;
  if (zk_client_.GetChildren(version_path.c_str(), &shards) != 0) {
    LOG_ERROR("UidCometServerMap::create --> get shards failed. path:" << version_path);
    return "";
  }

  std::string shard = boost::lexical_cast<std::string>(uid%(shards.size()));
  std::string shard_path = version_path + "/shard_" + shard;
  set(uid, shard_path);
  LOG_INFO("UidConnServerMap::create --> get endpoints successfully. path:" << shard_path << " uid:" << uid);
  return shard_path;
}

void UidConnServerMap::remove(int64_t uid, const std::string& conn_type) {
  LOG_INFO("UidConnServerMap::remove --> uid:" << uid);
  std::string cmd;
  if (conn_type == "sock") {
    cmd = "DEL " + boost::lexical_cast<std::string>(uid) + ":sock_server_key";
  } else if (conn_type == "comet") {
    cmd = "DEL " + boost::lexical_cast<std::string>(uid) + ":comet_server_key";
  } else {
    return;
  }
  redisReply * reply = NULL;
  bool res = false;
  for (int i = 0; i < 2; ++i) {
    reply = redisExecutor.Execute("redis server_key", cmd.c_str());
    if (reply != NULL) {
      if (reply->type == REDIS_REPLY_INTEGER) {
        res = true;
      }
      freeReplyObject(reply);
      if (res) {
        break;
      }
    }
  }
  if (!res) {
    LOG_ERROR("UidConnServerMap::remove --> redis del failed. uid:" << uid);
  }
}

std::string UidConnServerMap::get(int64_t uid, const std::string& conn_type) {
  std::string res = "";
  std::string cmd;
  if (conn_type == "sock") {
    cmd = "GET " + boost::lexical_cast<std::string>(uid) + ":sock_server_key";
  } else if (conn_type == "comet") {
    cmd = "GET " + boost::lexical_cast<std::string>(uid) + ":comet_server_key";
  } else {
    return "";
  }
  redisReply * reply = NULL;
  bool redisRes = false;
  for (int i = 0; i < 2; ++i) {
    reply = redisExecutor.Execute("redis server_key", cmd.c_str());
    if (reply != NULL) {
      if (reply->type == REDIS_REPLY_STRING) {
        res.assign(reply->str, reply->len);
        if (res.find("nil") != std::string::npos) {
          res = "";
        }
        redisRes = true;
      }
      freeReplyObject(reply);
      if (redisRes) {
        break;
      }
    }
  }
  if (!redisRes) {
    LOG_ERROR("UidCometServerMap::get --> redis get failed. uid:" << uid);
  }
  return res;
}

void ConnectionManagerServiceHandler::GetCometServiceKeyByUids(std::map<int64_t, std::string> & _return, const std::set<int64_t> & uids) {
  std::ostringstream os;
  for (std::set<int64_t>::const_iterator it = uids.end(); it != uids.end(); ++it) {
    if (it != uids.begin()) {
      os << ",";
    }
    os << (*it);
  }
  LOG_INFO("ConnectionManagerServiceHandler::GetCometServiceKeyByUids --> uids:" << os.str());
  BOOST_AUTO(it, uids.begin());
  for (; it != uids.end(); ++it) {
    string server_key = UidConnServerMap::instance().get(*it, "comet");
    if (!server_key.empty()) {
      _return.insert(make_pair(*it, server_key));
    }
  }
}

void ConnectionManagerServiceHandler::RemoveCometServiceKeyByUids(const std::set<int64_t>& uids) {
  std::ostringstream os;
  for (std::set<int64_t>::const_iterator it = uids.end(); it != uids.end(); ++it) {
    if (it != uids.begin()) {
      os << ",";
    }
    os << (*it);
  }
  LOG_INFO("ConnectionManagerServiceHandler::RemoveCometServiceKeyByUids --> uids:" << os.str());
  std::map<int64_t, std::string> res;
  BOOST_AUTO(it, uids.begin());
  for (; it != uids.end(); ++it) {
    UidConnServerMap::instance().remove(*it, "comet");
  }
}

void ConnectionManagerServiceHandler::AllocateCometServiceForUid(std::string& _return, const int64_t uid) {
  LOG_INFO("ConnectionManagerServiceHandler::AllocateCometServiceForUid --> uid:" << uid);
  _return = UidConnServerMap::instance().alloc(uid, "comet");
}

void ConnectionManagerServiceHandler::SetValidCometServiceForUid(int64_t uid, const std::string& server_key) {
  LOG_INFO("ConnectionManagerServiceHandler::SetValidCometServiceForUid --> uid:" << uid << " server_key:" << server_key);
  UidConnServerMap::instance().setValid(uid, server_key);
}

void ConnectionManagerServiceHandler::GetSockServiceKeyByUids(std::map<int64_t, std::string> & _return, const std::set<int64_t> & uids) {
  std::ostringstream os;
  for (std::set<int64_t>::const_iterator it = uids.end(); it != uids.end(); ++it) {
    if (it != uids.begin()) {
      os << ",";
    }
    os << (*it);
  }
  LOG_INFO("ConnectionManagerServiceHandler::GetSockServiceKeyByUids --> uids:" << os.str());
  BOOST_AUTO(it, uids.begin());
  for (; it != uids.end(); ++it) {
    string server_key = UidConnServerMap::instance().get(*it, "sock");
    if (!server_key.empty()) {
      _return.insert(make_pair(*it, server_key));
    }
  }
}

void ConnectionManagerServiceHandler::RemoveSockServiceKeyByUids(const std::set<int64_t>& uids) {
  std::ostringstream os;
  for (std::set<int64_t>::const_iterator it = uids.begin(); it != uids.end(); ++it) {
    if (it != uids.begin()) {
      os << ",";
    }
    os << (*it);
  }
  LOG_INFO("ConnectionManagerServiceHandler::RemoveSockServiceKeyByUids --> uids:" << os.str());
  std::map<int64_t, std::string> res;
  BOOST_AUTO(it, uids.begin());
  for (; it != uids.end(); ++it) {
    UidConnServerMap::instance().remove(*it, "sock");
  }
}

void ConnectionManagerServiceHandler::AllocateSockServiceForUid(std::string& _return, const int64_t uid) {
  LOG_INFO("ConnectionManagerServiceHandler::AllocateSockServiceForUid --> uid:" << uid);
  std::string server_key = UidConnServerMap::instance().get(uid, "sock");
  if (!server_key.empty() && server_key.find(":ok") != std::string::npos) {
    std::string sk = server_key.substr(0, server_key.find(":ok"));
    SockMessageOutletAdapter::instance().KickCurrentConn(sk, uid);
  }
  server_key = UidConnServerMap::instance().alloc(uid, "sock");
  SockMessageOutletAdapter::instance().GetOuterIPAndPort(server_key, _return);
}

void ConnectionManagerServiceHandler::SetValidSockServiceForUid(int64_t uid, const std::string& server_key) {
  LOG_INFO("ConnectionManagerServiceHandler::SetValidSockServiceForUid --> uid:" << uid << " server_key:" << server_key);
  UidConnServerMap::instance().setValid(uid, server_key);
}

}

